package org.jetlinks.pro.things.impl.connector;

import io.vavr.Function3;
import lombok.AllArgsConstructor;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.CollectionUtils;
import org.jetlinks.core.message.MessageType;
import org.jetlinks.core.message.ThingMessage;
import org.jetlinks.core.message.event.DefaultEventMessage;
import org.jetlinks.core.message.event.ThingEventMessage;
import org.jetlinks.core.message.property.DefaultReportPropertyMessage;
import org.jetlinks.core.message.property.ThingReportPropertyMessage;
import org.jetlinks.core.things.MetadataId;
import org.jetlinks.core.things.ThingId;
import org.jetlinks.core.things.ThingMetadataType;
import org.jetlinks.pro.streaming.Streaming;
import org.jetlinks.pro.things.ThingConstants;
import org.jetlinks.pro.things.ThingsConnector;
import org.jetlinks.pro.things.utils.ThingMetadataHelper;
import reactor.core.Disposable;
import reactor.core.Disposables;
import reactor.core.publisher.Mono;

import javax.annotation.Nonnull;
import javax.annotation.Nullable;
import java.util.Collections;
import java.util.Set;
import java.util.function.Function;

/**
 * 对源数据进行计算后返回新的物消息,支持将计算结果转为属性或者事件
 *
 * @author zhouhao
 * @see DefaultReportPropertyMessage
 * @see DefaultEventMessage
 * @since 1.12
 */
@Slf4j
@AllArgsConstructor
public class ComputeThingsConnector extends AbstractThingsConnector {

    /**
     * 用户触发当前连机器的数据源，一般为递归检查物连接器包裹的组合连接器
     * @see CompositeThingsConnector
     * @see org.jetlinks.pro.things.impl.RecursiveCheckThingsConnector
     */
    private final ThingsConnector source;

    //计算器,用于计算新的属性或事件结果
    private final Streaming<ThingMessage, ComputeResult, ComputeResult> computer;

    //连接到当前物的物模型
    @Override
    public Disposable doConnectTo(ThingId thingId,
                                MetadataId metadataId,
                                Function<ThingMessage, Mono<Void>> handler) {
        return doConnect(thingId, metadataId, source::connectTo, handler);
    }

    //连接到指定物的物模型
    @Override
    public Disposable doConnectTo(MetadataId metadataId, Function<ThingMessage, Mono<Void>> handler) {
        return doConnect(null, metadataId,
                         (ignore, mid, sHandler) -> source.connectTo(mid, sHandler),
                         handler);
    }


    private Disposable doConnect(@Nullable ThingId thingId,
                                 @Nonnull MetadataId metadataId,
                                 @Nonnull Function3<ThingId, MetadataId, Function<ThingMessage, Mono<Void>>, Disposable> source,
                                 @Nonnull Function<ThingMessage, Mono<Void>> handler) {
        Disposable.Composite disposable = Disposables.composite();

        //根据物模型类型转换为对应的物消息
        Function<ComputeResult, ThingMessage> converter = createConverter(thingId, metadataId);

        /**
         * 构造计算处理函数。使用 {@link org.jetlinks.pro.streaming.Computer#compute(Object)}计算后做处理
         */
        Function<ComputeResult, Mono<Void>> computeHandler = (result) -> {
            ThingMessage message = converter.apply(result);
            if(CollectionUtils.isNotEmpty(result.traces)){
                //添加连接器跟踪header，通常用于处理连接器是否存在递归
                message.getOrAddHeaderDefault(ThingConstants.connectorTrace)
                       .addAll(result.traces);
            }
            return handler.apply(message);
        };

        //从数据源获取数据并计算
        disposable.add(
            source
                .apply(thingId, metadataId, msg -> computer
                    .compute(msg)
                    .flatMap(computeHandler))
        );

        //订阅流式计算结果,比如使用了窗口时
        disposable.add(
            computer.output()
                    .flatMap(computeHandler)
                    .subscribe()
        );

        return disposable;
    }

    /**
     * 根据物模型类型转换为对应的物消息
     *
     * @param thingId 物ID
     * @param metadataId 物模型ID
     */
    private Function<ComputeResult, ThingMessage> createConverter(@Nullable ThingId thingId, MetadataId metadataId) {
        return ThingMetadataHelper
            .<ComputeResult, ThingMessage>create()
            //连接到属性时,转为上报属性消息
            .when(ThingMetadataType.property, (tid, mId, result) -> {
                ThingId sourceId = result.getThingId();
                Object data = result.getResult();

                return MessageType.REPORT_PROPERTY
                    //添加消息的物ID，当传入物ID为空时，使用计算结果中的物ID（连接到当前物模型时，物ID通常为空）
                    .<ThingReportPropertyMessage>forThing(tid == null ? sourceId : tid)
                    .success(Collections.singletonMap(mId.getId(), data));
            })
            //连接到事件时转为事件消息
            .when(ThingMetadataType.event, (tid, mId, result) -> {
                ThingId sourceId = result.getThingId();
                Object data = result.getResult();

                return MessageType.EVENT
                    //添加消息的物ID，当传入物ID为空时，使用计算结果中的物ID（连接到当前物模型时，物ID通常为空）
                    .<ThingEventMessage>forThing(tid == null ? sourceId : tid)
                    .event(mId.getId())
                    .data(data);
            })
            .toFunction(thingId, metadataId);
    }

    @AllArgsConstructor(staticName = "of")
    @Getter
    public static class ComputeResult {

        //物ID
        private ThingId thingId;

        //连接器跟踪标识。通常通过消息header获取
        private Set<String> traces;

        //计算结果
        private Object result;
    }

    @Override
    protected void onDispose() {
        source.dispose();
        computer.dispose();
    }

}
